# ******************************************************************#
##author: v_sunao3
##create time: 2023-10-26 14:02:50
# ******************************************************************#


# !/usr/bin/env python3
import sys
import time
from datetime import datetime
from pyhive import hive
import pymysql
import numpy as np

# 数据时间通过变量传入
yesterday = '${yyyy-MM-dd-1d}'
print("********************业务日期为：" + str(yesterday))
# 获取当前时间戳
current_timestamp = int(time.time())
# 获取当前时间
current_datetime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print("********************当前任务运行时间：" + str(current_datetime))


# 获取hive连接
def conn_to_hive(hive_database,
                 hive_host='172.25.1.64',
                 hive_port=7001,
                 hive_user='v_sunao3'
                 ):
    return hive.Connection(
        host=hive_host,
        port=hive_port,
        username=hive_user,
        database=hive_database
    )


# 连接mysql数据库
def conn_to_mysql(mysql_database,
                  mysql_host='172.25.1.58',
                  mysql_port=3306,
                  mysql_user='oceanus_user',
                  mysql_password='sqYcvrHc5gpEBH2a',
                  mysql_charset='utf8mb4',
                  ):
    return pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_user,
        password=mysql_password,
        database=mysql_database,
        charset=mysql_charset
    )


# 连接starrocks数据库
def conn_to_starrocks(starrocks_host='172.25.1.67',
                      starrocks_port=9030,
                      starrocks_user='test',
                      starrocks_password='srLcHLyUjUrvFf6R',
                      starrocks_database='test',
                      starrocks_charset='utf8mb4'
                      ):
    return pymysql.connect(
        host=starrocks_host,
        port=starrocks_port,
        user=starrocks_user,
        password=starrocks_password,
        database=starrocks_database,
        charset=starrocks_charset
    )


def split_hive_create_sql_str(create_sql_str):
    # 提取字段
    fields_info_main = create_sql_str.split("`(\n  `")[1].split("PARTITIONED BY (")[0].split(")\n")[0]

    result = []
    fields_info = fields_info_main.split(", \n")
    for field_info in fields_info:
        field_info_split = field_info.strip().split(" ")
        field_name = field_info_split[0].strip()
        field_type = field_info_split[1].strip()
        field_comment = field_info_split[3].strip()
        result.append([field_name, field_type, field_comment])
    return result


def get_hive_table_fields(hive_table, cursor_hive):
    '''
    获取hive表的字段与类型
    :param hive_table:
    :param cursor_hive:
    :return: hive表的字段与类型
    '''
    result = None
    try:
        # 获取Hive字段
        sql = f"show create table {hive_table}"
        cursor_hive.execute(sql)
        lines = cursor_hive.fetchall()
        create_sql_str = "\n".join([line[0] for line in lines])
        # todo 分割
        result = split_hive_create_sql_str(create_sql_str)

    except Exception as e:
        print("=" * 100)
        print(f"Failed to connect to Hive:{hive_table}, {str(e)}")
        print("=" * 100)
    finally:
        cursor_hive.close()

    return result


def split_mysql_create_sql_str(create_sql_str):
    # 提取字段
    fields_info_main = create_sql_str.split("` (\n  `")[1].split(") ENGINE=")[0]

    result = []
    fields_info = fields_info_main.split(",\n")
    for field_info in fields_info:
        if "PRIMARY KEY" in field_info: break
        if "COMMENT" in fields_info:
            field_info_split = field_info.strip().split("COMMENT")
            field_info_split_pre = field_info_split[0].split(" ")
            field_name = field_info_split_pre[0].strip()
            field_type = field_info_split_pre[1].strip()
            field_comment = field_info_split[1].strip()
        else:
            field_info_split = field_info.strip().split(" ")
            field_name = field_info_split[0].strip()
            field_type = field_info_split[1].strip()
            field_comment = ""

        result.append([field_name.lower(), field_type, field_comment])
    return result


def get_mysql_table_fields(mysql_table, cursor_mysql):
    result = None
    try:
        # 获取mysql表的数据量
        cursor_mysql.execute(f"show create table {mysql_table}")
        result = cursor_mysql.fetchone()
        create_sql_str = result[1]
        # todo 分割
        result = split_mysql_create_sql_str(create_sql_str)

    except Exception as e:
        print("=" * 100)
        print(f"Failed to connect to Mysql:{mysql_table} {str(e)}")
        print("=" * 100)
    finally:
        cursor_mysql.close()

    return result


def check_hive_mysql_fields(hive_table, mysql_table, conn_mysql, conn_hive):
    # 计算hive表数据量
    cursor_mysql = conn_mysql.cursor()
    cursor_hive = conn_hive.cursor()
    hive_table_fields_info = get_hive_table_fields(cursor_hive=cursor_hive, hive_table=hive_table)
    mysql_table_fields_info = get_mysql_table_fields(cursor_mysql=cursor_mysql, mysql_table=mysql_table)
    # 未同步
    if hive_table_fields_info is None:
        result = ['${yyyyMMdd-1d}', mysql_table, hive_table, 0,
                  "null", "null",
                  0, "null", current_datetime]
    else:
        # 已同步
        hive_table_fields_info_arr = np.array(hive_table_fields_info)
        mysql_table_fields_info_arr = np.array(mysql_table_fields_info)

        mysql_fields = mysql_table_fields_info_arr[:, 0]
        hive_fields = hive_table_fields_info_arr[:, 0]



        # 判断字段个数、字段名字、只要有一个不一致即为不一致
        hive_field_num = len(hive_fields)
        mysql_field_num = len(mysql_fields)
        script_run_time = current_datetime

        if hive_field_num == mysql_field_num :
            field_is_consistent = 1
            is_pass = "通过"
            for fields in zip(mysql_fields,hive_fields):
                mysql_field, hive_field = fields
                if mysql_field != hive_field:
                    field_is_consistent = 0
                    is_pass = "不通过"
                    break

        else:
            field_is_consistent = 0
            is_pass = "不通过"

        result = ['${yyyyMMdd-1d}', mysql_table, hive_table, 1,
                  hive_field_num, mysql_field_num,
                  field_is_consistent, is_pass, script_run_time]

    return result


def get_mysql_tables(cursor_mysql):
    try:
        # 获取mysql当前库的所有表
        cursor_mysql.execute("show tables")
        return cursor_mysql.fetchall()

    except Exception as e:
        print(f"Failed to connect to Mysql: {str(e)}")
        return None
    finally:
        cursor_mysql.close()


if __name__ == "__main__":

    hive_database = '${hive_database}'
    mysql_database = '${mysql_database}'

    # 链接数据库并创建游标
    conn_hive = conn_to_hive(hive_database=hive_database)
    conn_mysql = conn_to_mysql(mysql_database=mysql_database)
    conn_sr = conn_to_starrocks()

    mysql_tables = get_mysql_tables(conn_mysql.cursor())

    table_count = len(list(mysql_tables))
    if mysql_tables is not None:
        for mysql_table in list(mysql_tables):
            hive_table = f'ods_sg_zl_{mysql_table[0]}_df'
            result = check_hive_mysql_fields(hive_table=hive_table,
                                             mysql_table=mysql_table[0],
                                             conn_mysql=conn_mysql,
                                             conn_hive=conn_hive)

            # 插入数据
            cursor_sr = conn_sr.cursor()
            # print(result)
            # ['20231025', 'attachment', 'ods_sg_zl_attachment_df', 1, 16, 16, 1, '通过', '2023-10-26 15:53:03']
            try:
                '''
                `event_date` varchar(255) NOT NULL COMMENT "业务数据日期",
                `mysql_table_name` varchar(255) NOT NULL COMMENT "mysql表名",
                `hive_table_name` varchar(255) NOT NULL COMMENT "hive表名",
                `mysql2hive_status` tinyint(4) NULL COMMENT "mysql2hive同步状态， 0 未同步 1 已同步",
                `hive_field_num` bigint(20) NULL COMMENT "hive表字段个数",
                `mysql_field_num` bigint(20) NULL COMMENT "msyql表字段个数",
                `field_is_consistent` tinyint(4) NULL COMMENT "字段是否一致 0 不一致 1 一致",
                `is_pass` varchar(32) NULL COMMENT "是否通过 通过 和 不通过",
                `script_run_time` varchar(255) NULL COMMENT "脚本运行时间"
                '''
                event_date, mysql_table_name, hive_table_name, mysql2hive_status, hive_field_num, mysql_field_num, field_is_consistent, is_pass, script_run_time = result

                cursor_sr.execute(f"""
                                SELECT event_date from python_mysql_hive_ods_field_check_df 
                                where mysql_table_name='{mysql_table_name}' AND event_date='{event_date}' """)
                if cursor_sr.fetchone() is None:
                    sql = f"""INSERT into python_mysql_hive_ods_field_check_df 
                                        (event_date, mysql_table_name, hive_table_name, mysql2hive_status, hive_field_num, 
                                        mysql_field_num, field_is_consistent, is_pass, script_run_time) values
                                        ("{event_date}","{mysql_table_name}","{hive_table_name}",{mysql2hive_status},
                                        {hive_field_num},{mysql_field_num},"{field_is_consistent}","{is_pass}","{script_run_time}")
                                        """
                else:
                    sql = f"""UPDATE python_mysql_hive_ods_field_check_df set 
                                    hive_field_num={hive_field_num},
                                    mysql_field_num={mysql_field_num},
                                    mysql2hive_status={mysql2hive_status},
                                    field_is_consistent="{field_is_consistent}",
                                    script_run_time="{script_run_time}",
                                    is_pass="{is_pass}"
                                    where  mysql_table_name="{mysql_table_name}" AND event_date="{event_date}"
                                    """
                print("********************", sql)
                cursor_sr.execute(sql)
                conn_sr.commit()
                print('********************Data inserted successfully.')

            except Exception as e:
                conn_sr.rollback()
                print('********************Failed to insert data:', str(e))
            finally:

                # 关闭游标和连接
                cursor_sr.close()


    conn_mysql.close()
    conn_hive.close()
    conn_sr.close()
